package com.cloud.message;

import cn.hutool.core.util.RandomUtil;
import org.apache.commons.lang3.ObjectUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Mqtt 订阅抽象基类
 *
 * @param
 */
public class AbstractQTConsumer implements MqttCallback {

    private static Logger logger = LoggerFactory.getLogger(AbstractQTConsumer.class);

    private Object qtBean = null;
    private MqttClient mqttConsumer = null;
    private MqttConnectOptions options = null;
    private String hostUrl = null;
    private String clientId = null;
    private String topic = null;
    private int qos = 1;

    /**
     * 设置MqttClient客户端连接参数
     *
     * @param clientKey
     * @param hostUrl
     */
    public void setClientConnect(String hostUrl, String clientKey, String topic, int qos) {
        this.hostUrl = hostUrl;
        this.clientId = clientKey + "_" + RandomUtil.randomNumbers(6);
        this.topic = topic;
        this.qos = qos;
    }

    /**
     * 重新设置MqttClient对象
     */
    public MqttClient getMqttClient(MqttConnectOptions options, Object qtBean) {
        try {
            // 创建Mqtt客户端
            mqttConsumer = new MqttClient(hostUrl, clientId, new MemoryPersistence());
            // 建立Mqtt客户端连接
            mqttConsumer.connect(options);
            this.options = options;
            this.qtBean = qtBean;
        } catch (MqttException e) {
            e.printStackTrace();
        }
        return mqttConsumer;
    }

    /**
     * 消息到达后处理方法,可以自定义序列化和返回消费成功的相关逻辑
     *
     * @param topic   消息主题
     * @param message 上下文
     * @return 处理结果
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) {
        logger.debug("MQTT接收主题:{},级别:{}", topic, message.getQos());
    }

    /**
     * 连接断开后触发此方法
     *
     * @param cause
     */
    @Override
    public void connectionLost(Throwable cause) {
        // 连接断开
        logger.info("MQTT连接断开,30S之后尝试重连...");
        try {
            Thread.sleep(30000);
            AbstractQTMessage qtPushConsumer = (AbstractQTMessage) qtBean;
            if (ObjectUtils.isEmpty(mqttConsumer) || !mqttConsumer.isConnected()) {
                mqttConsumer.connect(options);
            }
            mqttConsumer.setCallback(qtPushConsumer);
            // 重新订阅主题
            mqttConsumer.subscribe(topic, qos, (String topic, MqttMessage message) -> qtPushConsumer.messageArrived(topic, message));
            logger.info("Mqtt消费者:" + mqttConsumer.getClientId() + "重连成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 消息到达后触发此方法
     *
     * @param token
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {

    }

}
